package com.zjc.springboot.delayqueue.listener;

import com.zjc.springboot.delayqueue.event.Msg;
import com.zjc.springboot.delayqueue.event.MsgEvent;
import com.zjc.springboot.delayqueue.redis.RedisClient;
import com.zjc.springboot.delayqueue.utils.HashUtils;
import com.zjc.springboot.zk.slice.ZookeeperClient;
import com.zjc.springboot.zk.slice.model.NodeInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

public class RedisExpiredListener implements MessageListener {

    private static final Logger logger = LoggerFactory.getLogger(RedisExpiredListener.class);

    ApplicationContext applicationContext;
    ZookeeperClient zookeeperClient;
    RedisClient redisClient;

    public RedisExpiredListener(ApplicationContext applicationContext, ZookeeperClient zookeeperClient, RedisClient redisClient) {

        this.applicationContext = applicationContext;
        this.zookeeperClient = zookeeperClient;
        this.redisClient = redisClient;
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String expiredKey = message.toString();
        if (expiredKey.startsWith(RedisClient.PREFIX_KEY)) {
            expiredKey = expiredKey.substring(RedisClient.PREFIX_KEY.length());

            logger.info("expiredKey :" + expiredKey);
            NodeInfo nodeInfo = zookeeperClient.getNodeInfo();
            int size = nodeInfo.getNodeSize();
            int id = nodeInfo.getNodeId();
            if (HashUtils.strHash(expiredKey, size) == id) {
                Msg msg = redisClient.getMsg(expiredKey);
                redisClient.deleteMsg(expiredKey);
                applicationContext.publishEvent(new MsgEvent(msg));
            }
        }
    }

}
